package broadcast

import (
	"encoding/hex"
	"runtime"

	"github.com/golang/protobuf/proto"
	"github.com/libp2p/go-libp2p-core/network"
	"hundsun.com/hsl/hschain/p2p/common/constant"
	"hundsun.com/hsl/hschain/p2p/impl/base"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node"
	"hundsun.com/hsl/hschain/p2p/impl/dht/protocol"
	ltypes "hundsun.com/hsl/hschain/p2p/impl/dht/protocol/broadcast/types"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
)

//V1 BroadcastV1
type V1 struct {
	*Protocol
}

//NewBroadcastV1 new broadcast V1
func NewBroadcastV1(b *Protocol) *V1 {
	return &V1{b}
}

//Init handlerV1初始化
func (b *V1) Init() {
	//注册p2p通信协议，用于处理节点之间请求
	//todo 封装p2pMsg
	protocol.RegisterStreamHandler(b.P2pNode.Host, constant.BroadcastBlockAddress, b.handleStreamBroadcastBlock)
	protocol.RegisterStreamHandler(b.P2pNode.Host, constant.BroadcastTransactionAddress, b.handleStreamBroadcastTx)

	// 处理event事件，向外节点广播消息
	txOutgoing := b.Comps.Sub(ltypes.TxTopic)
	blockOutgoing := b.Comps.Sub(ltypes.BlockTopic)
	cpu := runtime.NumCPU()
	for i := 0; i < cpu; i++ {
		go b.procBroadcastTxEvent(txOutgoing)
	}
	//区块广播
	go b.procBroadcastBlockEvent(blockOutgoing)
}

// 广播区块
func (b *V1) handleStreamBroadcastBlock(stream network.Stream) {
	// 1. 完成区块检查
	// 2. 完成区块广播
	// 3. 完成区块向缓存或者将区块给到共识模块
	blk := &pbledger.Block{}
	err := node.ReadStream(blk, stream)
	if err != nil {
		b.Log.Error("handleStreamBroadcastBlock", "ReadStream err", err)
		return
	}
	blkHash := hex.EncodeToString(blk.Hash())
	if b.BlockFilter.Contains(blkHash) {
		b.Log.Debug("handleStreamBroadcastBlock", "Repeat block hash", blkHash)
		return
	}
	// TODO 跟共识模块或者跟账本查找是否是重复区块，如果是着不进行处理, 如果不是则将区块发送到共识模块

	b.BlockFilter.Add(blkHash, struct{}{})
}

// 广播交易
func (b *V1) handleStreamBroadcastTx(stream network.Stream) {
	// 1. 完成tx交易检查
	// 2. 完成tx广播
	// 3. 完成tx向txpool存放
	tx := &pbledger.Transaction{}
	err := node.ReadStream(tx, stream)
	if err != nil {
		b.Log.Error("handleStreamBroadcastTx", "ReadStream err", err)
		return
	}
	txHash := hex.EncodeToString(tx.Hash())
	if b.TxFilter.Contains(txHash) {
		b.Log.Debug("handleStreamBroadcastTx", "Repeat tx hash", txHash)
		return
	}
	// TODO 跟共识模块或者跟账本查找是否是重复tx，如果是着不进行处理, 如果不是则将tx发送到txpool

	// 将消息广播出去
	opts := []base.MessageOption{
		base.WithFilters([]base.FilterStrategy{base.NearestStrategy}),
		base.WithChainID(b.Cfg.ChainID),
		// TODO 加入黑名单
		//base.WithBlackList(blacklist),
		//TODO 白名单可加入经过握手后的节点
		//base.WithWhiteList(whiteList),
	}
	go b.P2PEnv.SendMessages(tx, constant.BroadcastBlockAddress, opts...)
	b.BlockFilter.Add(txHash, struct{}{})
}

// 处理广播消息发布
func (b *V1) procBroadcastBlockEvent(out chan interface{}) {
	defer b.Comps.Unsub(out)
	for {
		select {
		case data, ok := <-out: //发送广播交易
			if !ok {
				return
			}
			msg, ok := data.(proto.Message)
			if ok {
				opts := []base.MessageOption{
					base.WithFilters([]base.FilterStrategy{base.NearestStrategy}),
					base.WithChainID(b.Cfg.ChainID),
					// TODO 加入黑名单
					//base.WithBlackList(blacklist),
					//TODO 白名单可加入经过握手后的节点
					//base.WithWhiteList(whiteList),
				}
				go b.P2PEnv.SendMessages(msg, constant.BroadcastBlockAddress, opts...)
			}
		case <-b.Ctx.Done():
			return
		}
	}
}

// 处理广播消息发布
func (b *V1) procBroadcastTxEvent(out chan interface{}) {
	defer b.Comps.Unsub(out)
	for {
		select {
		case data, ok := <-out: //发送广播交易
			if !ok {
				return
			}
			msg, ok := data.(proto.Message)
			if ok {
				opts := []base.MessageOption{
					base.WithFilters([]base.FilterStrategy{base.NearestStrategy}),
					base.WithChainID(b.Cfg.ChainID),
					// TODO 加入黑名单
					//base.WithBlackList(blacklist),
					//TODO 白名单可加入经过握手后的节点
					//base.WithWhiteList(whiteList),
				}
				go b.P2PEnv.SendMessages(msg, constant.BroadcastTransactionAddress, opts...)
			}

		case <-b.Ctx.Done():
			return
		}
	}
}
